package com.hbase.common;

import java.io.Closeable;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.NamespaceDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Append;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Increment;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Row;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.PageFilter;
import org.apache.hadoop.hbase.filter.PrefixFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.hbase.common.utils.ConfigUtil;

/**
 * hbasae基本操作
 * 
 * @author Administrator
 * 
 */
public class HBaseHelper implements Closeable {

	private static final Logger log = LoggerFactory.getLogger(HBaseHelper.class);
	// 设置每个表存储大小
	private long tableSize = ConfigUtil.getInt("config.properties", "hbase.put.tableSize", 10) * 1024 * 1024;
	// 是否写入日志记录
	private boolean setWriteToWAL = ConfigUtil.getBoolean("config.properties", "hbase.put.setWriteToWAL", false);
	// 默认一次查询最多2000条数据
	private static final int MAX_PAGE_SIZE = ConfigUtil.getInt("config.properties", "hbase.get.pageSize", 2000);
	// #本地缓存数据，当达到指定阈值，自动提交到hbase,默认是2M
	private static final int WRITE_BUFFER_SIZE = ConfigUtil.getInt("config.properties", "hbase.write.buffer", 2) * 1024 * 1024;

	private HBaseHelper() {
	}

	public static HBaseHelper getHelper() {
		return new HBaseHelper();
	}

	/**
	 * 创建命名空间
	 * 
	 * @param namespace
	 */
	public static void createNamespace(String namespace) {
		try {
			NamespaceDescriptor nd = NamespaceDescriptor.create(namespace).build();
			Admin admin = HbaseConnectionFactory.getConnection().getHBaseAdmin();
			admin.createNamespace(nd);
		} catch (Exception e) {
			System.err.println("Error: " + e.getMessage());
		}
	}

	/**
	 * 删除命名空间
	 * 
	 * @param namespace
	 * @param force
	 *            ：是否禁用表在删除，true-禁用
	 */
	public static void dropNamespace(String namespace, boolean force) {

		try {
			if (force) {
				Admin admin = HbaseConnectionFactory.getConnection().getHBaseAdmin();
				TableName[] tableNames = admin.listTableNamesByNamespace(namespace);
				for (TableName name : tableNames) {
					admin.disableTable(name);
					admin.deleteTable(name);
				}
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
		try {
			Admin admin = HbaseConnectionFactory.getConnection().getHBaseAdmin();
			admin.deleteNamespace(namespace);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	/**
	 * 获取所有表名
	 * 
	 * @return 返回表名集合
	 * @throws Exception
	 * @throws IOException
	 */
	public static List<String> getTableNames() throws IOException {
		List<String> list = new ArrayList<>();
		try {
			HBaseAdmin admin = HbaseConnectionFactory.getConnection().getHBaseAdmin();
			String[] tableNames = admin.getTableNames();
			if (tableNames != null && tableNames.length > 0) {
				for (String tableName : tableNames) {
					list.add(tableName);
				}
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
		return list;
	}

	/**
	 * 判断表是否存在
	 * 
	 * @param tableName
	 *            ：表名称
	 * @return
	 * @throws IOException
	 */
	public static boolean existsTable(String tableName) throws IOException {

		try {
			Admin admin = HbaseConnectionFactory.getConnection().getHBaseAdmin();
			return admin.tableExists(TableName.valueOf(tableName));
		} catch (Exception e) {
			e.printStackTrace();
		}
		return false;
	}

	/**
	 * 创建表
	 * 
	 * @param tableName
	 *            ：表名称
	 * @param familys
	 * @throws StorageException
	 */
	public static void createTable(String tableName, String... colfams) throws IOException {
		createTable(tableName, 1, colfams);
	}

	/**
	 * 
	 * @param tableName
	 *            ：表名称
	 * @param maxVersions
	 *            ：版本
	 * @param colfams
	 *            ：列族名称
	 * @throws IOException
	 */
	public static void createTable(String tableName, int maxVersions, String... colfams) throws IOException {

		try {
			HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(tableName));
			for (String cf : colfams) {
				HColumnDescriptor coldef = new HColumnDescriptor(cf);
				coldef.setMaxVersions(1);
				desc.addFamily(coldef);
			}
			Admin admin = HbaseConnectionFactory.getConnection().getHBaseAdmin();
			admin.createTable(desc);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	/**
	 * 
	 * @param tableName
	 *            ：表名称
	 * @param splitKeys
	 *            ：初始化region数据
	 * @param colfams
	 *            ：列族名称
	 * @throws IOException
	 */
	public static void createTable(String tableName, byte[][] splitKeys, String... colfams) throws IOException {

		try {
			HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(tableName));
			for (String cf : colfams) {
				HColumnDescriptor coldef = new HColumnDescriptor(cf);
				coldef.setMaxVersions(1);
				desc.addFamily(coldef);
			}
			Admin admin = HbaseConnectionFactory.getConnection().getHBaseAdmin();
			if (splitKeys != null) {
				admin.createTable(desc, splitKeys);
			} else {
				admin.createTable(desc);
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	/**
	 * 禁用表
	 * 
	 * @param tableName
	 * @throws IOException
	 */
	public static void disableTable(String tableName) throws IOException {

		try {
			Admin admin = HbaseConnectionFactory.getConnection().getHBaseAdmin();
			admin.disableTable(TableName.valueOf(tableName));
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	/**
	 * 启用表
	 * 
	 * @param tableName
	 * @throws IOException
	 */
	public static void enableTable(String tableName) throws IOException {

		try {
			Admin admin = HbaseConnectionFactory.getConnection().getHBaseAdmin();
			admin.enableTable(TableName.valueOf(tableName));
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	/**
	 * 删除表
	 * 
	 * @param tableName
	 *            ：表名称
	 * @throws IOException
	 */
	public static void dropTable(String tableName) throws IOException {

		try {
			Admin admin = HbaseConnectionFactory.getConnection().getHBaseAdmin();
			if (admin.isTableEnabled(TableName.valueOf(tableName))) {
				admin.disableTable(TableName.valueOf(tableName));
			}
			admin.deleteTable(TableName.valueOf(tableName));
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	/**
	 * 添加数据
	 * 
	 * @param tableName
	 *            ：表名称
	 * @param rowkey
	 *            ：主键
	 * @param family
	 *            ：列 族名称
	 * @param qualifier
	 *            ：列 名称
	 * @param value
	 *            ：列值
	 * @return
	 * @throws IOException
	 */
	public static void append(String tableName, String rowkey, String family, String qualifier, String value) throws IOException {

		Connection connection = HbaseConnectionFactory.getConnection().getHBaseConnection();
		HTable table = (HTable) connection.getTable(TableName.valueOf(tableName));
		Append append = new Append(Bytes.toBytes(rowkey));
		if (StringUtils.isNotBlank(family) && StringUtils.isNotBlank(qualifier) && StringUtils.isNotBlank(value)) {
			append.add(Bytes.toBytes(family), Bytes.toBytes(qualifier), Bytes.toBytes(value));
		}
		try {
			table.append(append);
			// table.setAutoFlush(false);
			// table.setWriteBufferSize(WRITE_BUFFER_SIZE);
		} catch (Exception e) {
			e.printStackTrace();
			table.flushCommits();
		}
	}

	/**
	 * 添加数据
	 * 
	 * @param tableName
	 *            ：表名称
	 * @param rowkey
	 *            ：主键
	 * @param family
	 *            ：列 族名称
	 * @param qualifier
	 *            ：列 名称
	 * @param value
	 *            ：列值
	 * @return
	 */
	public static void put(String tableName, String rowkey, String family, String qualifier, String value) throws IOException {
		put(tableName, rowkey, family, qualifier, System.currentTimeMillis(), value);
	}

	/**
	 * 添加数据
	 * 
	 * @param tableName
	 *            ：表名称
	 * @param rowkey
	 *            ：主键
	 * @param family
	 *            ：列 族名称
	 * @param qualifier
	 *            ：列 名称
	 * @param ts
	 *            :时间，毫秒数
	 * @param value
	 *            ：列值
	 * @return
	 */
	public static void put(String tableName, String rowkey, String family, String qualifier, long ts, String value) throws IOException {
		Put put = new Put(Bytes.toBytes(rowkey));
		put.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier), ts, Bytes.toBytes(value));
		put(tableName, put);
	}

	/**
	 * 添加数据
	 * 
	 * @param tableName
	 *            ：表名称
	 * @param puts
	 *            ：数据
	 * @return
	 */
	public static void put(String tableName, Put put) throws IOException {
		List<Put> puts = new ArrayList<Put>();
		puts.add(put);
		batchPut(tableName, puts);
	}

	/**
	 * 添加数据
	 * 
	 * @param tableName
	 *            ：表名称
	 * @param rowkey
	 *            ：主键
	 * @param family
	 *            ：列 族
	 * @param map
	 *            ：列名称及数据
	 * @return
	 */
	public static void put(String tableName, String rowkey, String family, Map<String, Object> map) throws IOException {

		try {
			List<Put> puts = new ArrayList<Put>();
			Put put = new Put(rowkey.getBytes());
			if (map != null && map.size() > 0) {
				for (String qualifier : map.keySet()) {
					Object object = map.get(qualifier);
					String value = (object == null) ? "" : object.toString();
					put.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier), Bytes.toBytes(value));
				}
			}
			puts.add(put);
			batchPut(tableName, puts);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	/**
	 * 批量添加数据
	 * 
	 * @param tableName
	 *            ：表名称
	 * @param puts
	 *            ：数据
	 * @return
	 * @throws StorageException
	 */
	public static void batchPut(String tableName, List<Put> puts) throws IOException {

		Connection connection = HbaseConnectionFactory.getConnection().getHBaseConnection();
		HTable table = (HTable) connection.getTable(TableName.valueOf(tableName));
		try {
			// 设置写缓存大小，当达到指定值，自动提交到服务；或者关闭表时，自动提交
			// table.setAutoFlush(false);
			// table.setWriteBufferSize(WRITE_BUFFER_SIZE);
			table.batch(puts);
		} catch (Exception e) {
			e.printStackTrace();
			table.flushCommits();
		}
	}

	/**
	 * 添加数据
	 * 
	 * @param tableName
	 *            ：表名称
	 * @param rowkey
	 *            ：主键
	 * @param family
	 *            ：列 族
	 * @param qualifier
	 *            ：列
	 * @param value
	 *            ：列名称数据
	 * @return
	 */
	public static void batchIncrement(String tableName, String rowKey, String family, String qualifier, long value) {

		try {
			Connection connection = HbaseConnectionFactory.getConnection().getHBaseConnection();
			Table table = connection.getTable(TableName.valueOf(tableName));
			table.incrementColumnValue(rowKey.getBytes(), family.getBytes(), qualifier.getBytes(), value);
			Increment increment = new Increment(Bytes.toBytes(rowKey));
			table.increment(increment);
			table.close();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	/**
	 * @desc: 删除数据
	 * @param tableName
	 *            ：表名称
	 * @param rowKey
	 *            :行健
	 * @param family
	 *            ：列族
	 * @throws IOException
	 */
	public static void delete(String tableName, String rowKey, String family) throws IOException {
		delete(tableName, rowKey, family, null);
	}

	/**
	 * @desc: 删除数据
	 * @param tableName
	 *            ：表名称
	 * @param rowKey
	 *            :行健
	 * @param family
	 *            ：列族
	 * @param qualifier
	 *            ：列明细
	 * @throws IOException
	 */
	public static void delete(String tableName, String rowKey, String family, String qualifier) throws IOException {

		delete(tableName, rowKey, family, qualifier, null);
	}

	/**
	 * @desc: 删除数据
	 * @param tableName
	 *            ：表名称
	 * @param rowKey
	 *            :行健
	 * @param family
	 *            ：列族
	 * @param qualifier
	 *            ：列明细
	 * @param timestamp
	 *            ：时间，毫秒数
	 * @throws IOException
	 */
	public static void delete(String tableName, String rowKey, String family, String qualifier, Long timestamp) throws IOException {

		try {
			// 创建删除一行数据
			Delete delete = new Delete(rowKey.getBytes());
			if (StringUtils.isNotBlank(family) && StringUtils.isNotBlank(qualifier) && timestamp != null) {
				// 添加删除的列数据
				delete.addColumn(family.getBytes(), qualifier.getBytes(), timestamp);
			} else if (StringUtils.isNotBlank(family) && StringUtils.isNotBlank(qualifier)) {
				// 添加删除的列数据
				delete.addColumn(family.getBytes(), qualifier.getBytes());
			} else if (StringUtils.isNotBlank(family)) {
				delete.addFamily(family.getBytes());
			}
			delete(tableName, delete);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	/**
	 * @desc: 删除数据
	 * @param tableName
	 *            ：表名称
	 * @param delete
	 *            :删除条件
	 * @throws IOException
	 */
	public static void delete(String tableName, Delete delete) throws IOException {

		try {
			Connection connection = HbaseConnectionFactory.getConnection().getHBaseConnection();
			Table table = connection.getTable(TableName.valueOf(tableName));
			// 添加删除数据
			table.delete(delete);
			table.close();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	/**
	 * @desc: 删除数据
	 * @param tableName
	 *            ：表名称
	 * @param list
	 *            :删除条件集
	 * @throws IOException
	 */
	public static void batchDelete(String tableName, List<Delete> list) throws IOException {

		try {
			Connection connection = HbaseConnectionFactory.getConnection().getHBaseConnection();
			Table table = connection.getTable(TableName.valueOf(tableName));
			// 添加删除数据
			table.delete(list);
			table.close();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	/**
	 * @Description:根据rowkey+表名查询,返回数据
	 * @param rowkey
	 *            ： rowkey
	 * @param tableName
	 *            表名
	 * @throws Exception
	 * @return List<Map<String,String>>
	 */
	public static Map<String, String> get(String tableName, String rowkey) throws IOException {

		Get get = new Get(rowkey.getBytes());
		Map<String, String> map = get(tableName, get, null);
		return map;
	}

	/**
	 * @desc:查询数据
	 * @param tableName
	 *            ：表名称
	 * @param get
	 *            ：行数据集
	 * @throws IOException
	 */
	public static Map<String, String> get(String tableName, Get get, Filter filter) throws IOException {

		Map<String, String> data = new HashMap<String, String>();
		try {
			Connection connection = HbaseConnectionFactory.getConnection().getHBaseConnection();
			// 创建表对象
			Table table = connection.getTable(TableName.valueOf(tableName));
			if (filter != null) {
				get.setFilter(filter);
			}
			Result rs = table.get(get);
			data = getResult(rs);
			table.close();
			return data;
		} catch (Exception e) {
			e.printStackTrace();
		}
		return data;
	}

	/**
	 * 获取查询结果集
	 */
	private static Map<String, String> getResult(Result rs) throws IOException {

		Map<String, String> data = new HashMap<String, String>();
		try {
			if (rs == null || rs.isEmpty()) {
				return data;
			}
			data.put("ROWKEY", Bytes.toString(rs.getRow()));
			for (Cell cell : rs.listCells()) {
				String key = Bytes.toString(CellUtil.cloneQualifier(cell));
				data.put(key, Bytes.toString(CellUtil.cloneValue(cell)));
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
		return data;
	}

	/**
	 * @desc:查询数据
	 * @param tableName
	 * @param gets
	 *            ：行数据集
	 * @throws IOException
	 */
	public static List<Map<String, String>> batchGet(String tableName, List<Get> gets) throws IOException {

		List<Map<String, String>> resultList = new ArrayList<Map<String, String>>();
		try {
			Connection connection = HbaseConnectionFactory.getConnection().getHBaseConnection();
			// 创建表对象
			Table table = connection.getTable(TableName.valueOf(tableName));
			Result[] rs = table.get(gets);
			if (rs == null) {
				return resultList;
			}
			for (Result result : rs) {
				Map<String, String> data = getResult(result);
				resultList.add(data);
			}
			table.close();
		} catch (Exception e) {
			e.printStackTrace();
		}
		return resultList;
	}

	/**
	 * 批量处理，支持put/delete/Get等操作
	 * 
	 * @param tableName
	 *            ：表名称
	 * @param list
	 *            ：数据
	 * @param results
	 *            :返回执行结果
	 * @throws IOException
	 */
	@SuppressWarnings("deprecation")
	public static Object[] batch(String tableName, List<Row> list) throws IOException {

		int size = (list != null && list.size() > 0) ? list.size() : 0;
		Object[] obj = new Object[size];
		try {
			Connection connection = HbaseConnectionFactory.getConnection().getHBaseConnection();
			Table table = connection.getTable(TableName.valueOf(tableName));
			obj = table.batch(list);
		} catch (Exception e) {
			e.printStackTrace();
		}
		return obj;
	}

	/**
	 * 批量处理，支持put/delete/Get等操作
	 * 
	 * @param tableName
	 *            ：表名称
	 * @param list
	 *            ：数据
	 * @param results
	 *            :返回执行结果
	 * @throws IOException
	 */
	public static void batch(String tableName, List<Row> list, Object[] results) throws IOException {
		
		try {
			Connection connection = HbaseConnectionFactory.getConnection().getHBaseConnection();
			Table table = connection.getTable(TableName.valueOf(tableName));
			table.batch(list, results);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	/**
	 * @Description: 根据rowkey前缀+表名模糊查询,返回数据(最多100条)
	 * @param tableName
	 *            : 表名称
	 * @param scan
	 *            : 查询条件
	 * @throws Exception
	 * @return List<Map<String,String>>
	 * @date 2018年1月31日
	 */
	private static List<Map<String, String>> getResult(String tableName, Scan scan) {

		List<Map<String, String>> dataList = new ArrayList<Map<String, String>>();
		try {

			// 设置scan扫描表缓存大小，可以在hbasae-site.xml中配置hbase.client.scanner.caching，scan扫描缓存级别高于表级别缓存
			// scan.setCaching(caching);
			// 设置批量处理大小
			// scan.setBatch(batch);
			Connection connection = HbaseConnectionFactory.getConnection().getHBaseConnection();
			HTable table = (HTable) connection.getTable(TableName.valueOf(tableName));
			// 设置表扫描缓存大小
			// table.setScannerCaching(scannerCaching);
			ResultScanner rs = table.getScanner(scan);
			if (rs == null) {
				return dataList;
			}
			for (Result result : rs) {
				Map<String, String> data = getResult(result);
				data.put("ROWKEY", Bytes.toString(result.getRow()));
				dataList.add(data);
			}
			table.close();
			rs.close();
		} catch (Exception e) {
			e.printStackTrace();
		}
		return dataList;
	}

	/**
	 * @Description: 根据rowkey前缀+表名模糊查询,返回数据(最多100条)
	 * @param rowkeyPrefix
	 *            : rowkey前缀
	 * @param tableName
	 *            : 表名
	 * @throws Exception
	 * @return List<Map<String,String>>
	 */
	public static List<Map<String, String>> scan(String tableName, String rowkeyPrefix, Filter filter) {

		List<Map<String, String>> dataList = new ArrayList<Map<String, String>>();
		try {
			Scan scan = new Scan();
			FilterList filters = new FilterList();
			filters.addFilter(new PageFilter(MAX_PAGE_SIZE)); // 分页返回数量
			if (StringUtils.isNotBlank(rowkeyPrefix)) {
				filters.addFilter(new PrefixFilter(rowkeyPrefix.getBytes()));
			}
			if (filter != null) {
				filters.addFilter(filter);
			}
			scan.setFilter(filters);

			// 模糊查询
			dataList = getResult(tableName, scan);
		} catch (Exception e) {
			e.printStackTrace();
		}
		return dataList;
	}

	/**
	 * 
	 * @desc:查询数据
	 * @param tableName
	 *            ：表名称
	 * @param rowkey
	 *            :行健
	 * @param family
	 *            :列族
	 * @param qualifier
	 *            :列明细
	 * @return
	 * @throws IOException
	 */
	public static List<Map<String, String>> scan(String tableName, String rowkey, String family, String qualifier) throws IOException {

		List<String> columns = new ArrayList<String>();
		if (StringUtils.isNotBlank(qualifier)) {
			columns.add(qualifier);
		}
		List<Map<String, String>> list = scan(tableName, rowkey, family, columns);
		return list;
	}

	/**
	 * 
	 * @desc:查询数据
	 * @param tableName
	 *            ：表名称
	 * @param rowkey
	 *            :行健
	 * @param family
	 *            :列族
	 * @param qualifiers
	 *            :列明细
	 * @return
	 * @throws IOException
	 */
	public static List<Map<String, String>> scan(String tableName, String rowkey, String family, List<String> qualifiers) throws IOException {

		List<Map<String, String>> dataList = new ArrayList<Map<String, String>>();
		try {
			Scan scan = null;
			if (StringUtils.isEmpty(rowkey)) {
				scan = new Scan();
			} else {
				scan = new Scan(rowkey.getBytes());
			}

			if (family != null) {
				scan.addFamily(family.getBytes());
				if (qualifiers != null && qualifiers.size() > 0) {
					for (String qualifier : qualifiers) {
						scan.addColumn(family.getBytes(), qualifier.getBytes());
					}
				}
			}
			scan.setFilter(new PageFilter(MAX_PAGE_SIZE));
			// 模糊查询
			dataList = getResult(tableName, scan);
		} catch (Exception e) {
			e.printStackTrace();
		}
		return dataList;
	}

	/**
	 * 
	 * @desc:查询数据
	 * @param tableName
	 *            ：表名称
	 * @param rowkey
	 *            :行健
	 * @param family
	 *            :列族
	 * @param columns
	 *            :key为列族:value值为列明细
	 * @return
	 * @throws IOException
	 */
	public static List<Map<String, String>> scan(String tableName, String rowkey, List<Map<String, String>> columns, Filter filter)
			throws IOException {

		List<Map<String, String>> dataList = new ArrayList<Map<String, String>>();
		try {
			Scan scan = null;
			if (StringUtils.isEmpty(rowkey)) {
				scan = new Scan();
			} else {
				scan = new Scan(rowkey.getBytes());
			}
			if (columns != null && columns.size() > 0) {
				for (Map<String, String> map : columns) {
					for (String family : map.keySet()) {
						if (StringUtils.isNotBlank(family) && StringUtils.isNotBlank(map.get(family))) {
							scan.addColumn(Bytes.toBytes(family), Bytes.toBytes(map.get(family)));
						} else if (StringUtils.isNotBlank(family)) {
							scan.addFamily(Bytes.toBytes(family));
						}
					}
				}
			}

			FilterList filters = new FilterList();
			filters.addFilter(new PageFilter(MAX_PAGE_SIZE)); // 分页返回数量
			if (filter != null) {
				filters.addFilter(filter);
			}
			scan.setFilter(filters);
			// 模糊查询
			dataList = getResult(tableName, scan);
		} catch (Exception e) {
			e.printStackTrace();
		}
		return dataList;
	}

	/**
	 * @Description:根据rowkey前缀+表名模糊查询,返回数据(最多100条)
	 * @param rowkeyPrefix
	 *            : rowkey前缀
	 * @param tableName
	 *            ： 表名
	 * @throws Exception
	 */
	public static List<Map<String, String>> scan(String tableName, String startRow, String stopRow, Filter filter) throws Exception {
		// 模糊查询
		List<Map<String, String>> dataList = scan(tableName, startRow, stopRow, null, filter);
		return dataList;
	}

	/**
	 * 
	 * @Description:分页查询
	 * @param tableName
	 *            ：表名称
	 * @param startRow
	 *            ：开始行
	 * @param stopRow
	 *            ：结束行
	 * @param columns
	 *            ：查询那些列
	 * @param reqFilter
	 *            ：过滤规则
	 * @return
	 * @return List<Map<String, String>>： 返回类型
	 * @throws
	 */
	public static List<Map<String, String>> scan(String tableName, String startRow, String stopRow, List<Map<String, String>> columns, Filter filter)
			throws Exception {

		List<Map<String, String>> dataList = new ArrayList<Map<String, String>>();
		try {
			Scan scan = new Scan();
			if (StringUtils.isNotBlank(startRow)) {
				scan.setStartRow(Bytes.toBytes(startRow));
			}
			if (StringUtils.isNotBlank(stopRow)) {
				scan.setStopRow(Bytes.toBytes(stopRow));
			}
			if (columns != null && columns.size() > 0) {
				for (Map<String, String> map : columns) {
					for (String family : map.keySet()) {
						if (StringUtils.isNotBlank(family) && StringUtils.isNotBlank(map.get(family))) {
							scan.addColumn(Bytes.toBytes(family), Bytes.toBytes(map.get(family)));
						} else if (StringUtils.isNotBlank(family)) {
							scan.addFamily(Bytes.toBytes(family));
						}
					}
				}
			}
			FilterList filters = new FilterList();
			filters.addFilter(new PageFilter(MAX_PAGE_SIZE)); // 分页返回数量
			if (filter != null) {
				filters.addFilter(filter);
			}
			scan.setFilter(filters);
			// scan.setBatch(100000);
			// 模糊查询
			dataList = getResult(tableName, scan);
		} catch (Exception e) {
			e.printStackTrace();
		}
		return dataList;
	}

	/**
	 * 
	 * @param tableName
	 *            :表名称
	 * @param scan
	 *            ：查询条件
	 * @param filter
	 *            ：过滤规则
	 * @return
	 * @throws Exception
	 */
	public static List<Map<String, String>> scan(String tableName, Scan scan) throws Exception {
		if (scan == null) {
			scan = new Scan();
		}
		List<Map<String, String>> dataList = getResult(tableName, scan);
		return dataList;
	}

	@Override
	public void close() throws IOException {
		Connection connection = HbaseConnectionFactory.getConnection().getHBaseConnection();
		connection.close();
	}

}
